﻿using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Transactions;
using eLibrary;
using eLibrary.Constant;
using eTask.Extensions;

namespace Letter.IO.Channels;

sealed partial class Channel<T> : IAsyncAbortable, IChannel<T>
{
    private readonly static TransactionAbortedException ABORTED_EXCEPTION = new TransactionAbortedException();
    private readonly static WaitCallback ABORT_TASK = state => { ((IAsyncAbortable)state).AbortAsync().Ignore(); };

    public Channel(bool allowSynchronousContinuations)
    {
        this.SourceChannel = Channel.CreateUnbounded<T>(new UnboundedChannelOptions()
        {
            SingleReader = true,
            SingleWriter = false,
            AllowSynchronousContinuations = allowSynchronousContinuations
        });

        this.Reader = new ChannelReader(this);
        this.Writer = new ChannelWriter(this);
        
        Interlocked.Exchange(ref this.count, 0);
        Interlocked.Exchange(ref this.writable, Booleans.TRUE);
        this.ReadLoopTask = this.ChannelReadLoop();
    }
    
    private long count = 0;
    private long writable = Booleans.TRUE;
    
    public long Pressure
    {
        get { return Interlocked.Read(ref this.count); }
    }
    
    public bool Writable
    {
        get { return Interlocked.Read(ref this.writable) == Booleans.TRUE; }
    }

    private ActionAsync<T> NotifyFunc
    {
        get;
        set;
    }
    
    private Task ReadLoopTask
    {
        get;
        set;
    }

    private System.Threading.Channels.Channel<T> SourceChannel
    {
        get; 
        set;
    }

    private System.Threading.Channels.ChannelReader<T> SourceReader
    {
        get { return this.SourceChannel.Reader; }
    }
    
    private System.Threading.Channels.ChannelWriter<T> SourceWriter
    {
        get { return this.SourceChannel.Writer; }
    }

    public IChannelReader<T> Reader
    {
        get;
    }

    public IChannelWriter<T> Writer
    {
        get;
    }
    
    public async ValueTask<bool> TryWriteAsync(T o)
    {
        if (o == null)
        {
            throw new ArgumentNullException(nameof(o));
        }

        if (this.Writable)
        {
            if (await this.SourceWriter.WaitToWriteAsync())
            {
                Interlocked.Increment(ref this.count);
                return this.SourceWriter.TryWrite(o);
            }
        }
        
        return false;
    }

    private async Task ChannelReadLoop()
    {
        try
        {
            while (true)
            {
                var result = await this.SourceReader.WaitToReadAsync();
                if (!result)
                {
                    break;
                }

                if (!this.SourceReader.TryRead(out var item))
                {
                    break;
                }

                await this.NotifyFunc(item);
                Interlocked.Decrement(ref this.count);
            }
        }
        catch (Exception e) when (e is TransactionAbortedException)
        {
        }
        catch (Exception e)
        {
            throw;
        }
    }

    public void Release()
    {
        Interlocked.Exchange(ref this.count, 0);
        Interlocked.Exchange(ref this.writable, Booleans.FALSE);
        ThreadPool.UnsafeQueueUserWorkItem(ABORT_TASK, this);
    }
    
    public async ValueTask AbortAsync()
    {
        this.SourceWriter.Complete(ABORTED_EXCEPTION);
        await this.ReadLoopTask;
    }
}

interface IAsyncAbortable
{
    ValueTask AbortAsync();
}
